package com.navinfo.platform.test;

import com.alibaba.fastjson.JSON;
import com.navinfo.opentsp.platform.location.protocol.common.LCIncidentAlarmData;
import com.navinfo.opentsp.platform.location.protocol.common.LCPredictiveMaintenanceD00B;
import com.navinfo.platform.bean.*;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class testKafkaClient_d00b {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "10.30.50.128:12892");//you should write specific ip address rather than localhost
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//StringSerializer/IntegerSerializer/or other self-defined Serializer.
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
//        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", "MyGroup");
        properties.put("auto.offset.reset","latest");
        final String splitStr = "\001";
        Consumer<String,byte[]> consumer  = new KafkaConsumer<String,byte[]>(properties);
        consumer.subscribe(Arrays.asList("predictRepairD00B"));

        while(true){
            ConsumerRecords<String,byte[]> records = consumer.poll(100);
            for(ConsumerRecord re:records){
                try {
                    LCPredictiveMaintenanceD00B.PredictiveMaintenanceD00B predictiveMaintenance = LCPredictiveMaintenanceD00B.PredictiveMaintenanceD00B.parseFrom((byte[]) (re.value()));
//                    LCIncidentAlarmData.IncidentAlarmData incidentAlarmData = LCIncidentAlarmData.IncidentAlarmData.parseFrom((byte[]) (re.value()));
                    StringBuilder predictiveMaintenanceStr = new StringBuilder(5000);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getLatitude()).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getLongitude()).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getOriginalLat()).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getOriginalLng()).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getHeight()).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getDirection()).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getGpsDate()).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getVersion()).append(splitStr);
                    predictiveMaintenanceStr.append(JSON.toJSON(MaintenanceEvent1D00BDto.parseString(predictiveMaintenance.getMaintenanceEvent1()))).append(splitStr);
                    predictiveMaintenanceStr.append(JSON.toJSON(MaintenanceEvent2D00BDto.parseString(predictiveMaintenance.getMaintenanceEvent2()))).append(splitStr);
                    predictiveMaintenanceStr.append(JSON.toJSON(MaintenanceEvent3D00BDto.parseString(predictiveMaintenance.getMaintenanceEvent3()))).append(splitStr);
                    predictiveMaintenanceStr.append(JSON.toJSON(MaintenanceEvent4D00BDto.parseString(predictiveMaintenance.getMaintenanceEvent4()))).append(splitStr);
                    predictiveMaintenanceStr.append(JSON.toJSON(MaintenanceEvent5D00BDto.parseString(predictiveMaintenance.getMaintenanceEvent5()))).append(splitStr);
                    predictiveMaintenanceStr.append(JSON.toJSON(MaintenanceEvent6D00BDto.parseString(predictiveMaintenance.getMaintenanceEvent6()))).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getTerminalId()).append(splitStr);
                    predictiveMaintenanceStr.append(predictiveMaintenance.getSerialNumber()).append(splitStr);
                    System.out.println(predictiveMaintenance.getTerminalId());

                }catch (Exception e){
                   e.printStackTrace();
                }
            }
        }
    }
}